package com.cmbyc.Kafka_Flink.main

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, tableConversions}
import org.apache.flink.table.descriptors.{Csv, FileSystem, OldCsv, Schema}

/**
 *
 * @program: com.cmbyc.Kafka_Flink.main
 * @author: YCLW058
 * @create: 2021-05-21 11:38
 * @decsription:
 *
 * */

object File_csv {
  def main(args: Array[String]): Unit = {
    //1 env
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val tenv = StreamTableEnvironment.create(env)

    //2 source
    val filePath="D:\\tool\\server\\IDEA\\code\\Java\\Flink\\Flink_scala\\data\\input\\test"
tenv.connect(new FileSystem().path(filePath))
  .withFormat(new Csv())
  .withSchema(new Schema()
  .field("id",DataTypes.INT())
  .field("name",DataTypes.STRING())
  .field("age",DataTypes.INT())
  )
  .createTemporaryTable("inputTable")
    //3 transformation
    //4 sink
    val resultTable = tenv.from("inputTable")
    resultTable.toAppendStream[(Int,String,Int)].print()
    //5 execute
    env.execute()
  }

}
